#!/usr/local/bin/perl -w
use strict;
use utf8;

use EV;
use HTTP::Parser::XS qw(parse_http_request);
use IO::Socket::INET;
use JSON::XS;
use Digest::MD5 qw(md5_hex);
require HTTP::Status;
use Time::HiRes qw(gettimeofday);

my $port = shift || 12345;
my $debug = shift;
my $ttl = 120;
binmode STDOUT,":utf8";

# список персональных сообщений для каждого клианта
# каналы: создание, удаление, подписка
# формат ответа
# type=>"response",command=>"login",error=>0
# error = 0 - нет ошибки



# список клиентов по логинам
my %logins;
# список клиентов по сессиям
my %sids;
# список клиентов ожидающих сообщений
my %waiting;
# все ожидающие чтения слушатели
my %readers;
# все ожидающие записи слушатели
my %writers;
# обработчики канальных списков
my %chanel_lister;

my %chanels;

my %command = (
	login => \&cmd_login,
	check => \&cmd_check,
	message => \&cmd_message,
	private => \&cmd_private,
	join => \&cmd_join_chanel,
	leave => \&cmd_leave_chanel,
	create => \&cmd_create_chanel,
	remove => \&cmd_remove_chanel,
	list => \&cmd_list_chanel,
);

# серверный сокет
my $lsn = new IO::Socket::INET(
	    Listen => 100,
	    LocalHost => "127.0.0.1",
	    LocalPort => $port,
	    ReuseAddr => 1,
	);

# слушаем основной сокет
my $LoopWaiter = EV::io $lsn, EV::READ, \&onstream_handler;

add_chanel("","main");
add_chanel("","trade", nolist=>1);

EV::loop;

################################################
sub check_auth {
	my $sid = shift;
	unless($sid){
		return undef;
	}elsif(my $s = $sids{$sid}){
		my $now = time();
		if($s->{last} + $ttl < $now){
			if(my $lgn = $logins{$s->{login}}){
				if($lgn->{sid} eq $sid){
					clear_session($s->{login});
				}
			}
			delete $sids{$sid};
			return undef;
		}else{
			$s->{last} = $now;
			return $s->{login};
		}
	}else{
		return undef;
	}
}
sub ts {
	return sprintf "%02d:%02d:%02d",(localtime)[2,1,0];
}
sub ats {
	return time();
}
sub onstream_handler {
	# принимаем подключение
	if($debug){
		print "accept\n";
	}
    my ($watcher, $revents) = @_;
    my $new = $lsn->accept;
    unless($new){
		if($debug){
			print "accept failed: $!\n";
		}
		return 0;
    }
    my $nw = EV::io $new, EV::READ, \&onread_handler;
	# сохраняем слушателя, сокет и буфер с принятыми данными
    $readers{int $nw} = [$nw,$new,""];
}
sub onread_handler {
	# Получили данные
	my ($w, $r) = @_;
	my $ary = $readers{int $w};
	my $buf;
	$ary->[1]->recv($buf,1024);
	if($debug){
		my $buf2 = $buf;
		#utf8::decode($buf2);
		print "read: $buf2\n";
	}
	if($buf){
		# есть данные, цепляем к предыдущим если есть
		unless(process($ary->[1], $buf = ($ary->[2] .= $buf))){
			# сообщение не полное, ждем дальше
			return 0;
		}
	}
	# раобота с содинением закончена, закрываем и чистим
	delete $readers{int $w};
	# сокет должен сам закрытся, если его не сохранили в process(...)
}

sub process {
	my ($fh,$str) = @_;
	if(length($str) > 4096){
		# очень длинные запросы не обрабатываем
		send_message($fh,{status=>200,type=>"response",command=>"",error=>413,text=>"too long message",ts=>ts(),rid => -1});
		return 1;
	}
	my $param = {};
	my $header_size = parse_http_request($str,$param);
	if($header_size == -2){
		# неполное сообщение
		return undef;
	}elsif($header_size == -1){
		# поломанное сообщение
		send_message($fh,{status=>200,type=>"response",command=>"",error=>400,text=>"broken message",ts=>ts(),rid => -1});
		return 1;
	}
	my $body = substr($str,$header_size);
	if($param->{CONTENT_LENGTH} && $param->{CONTENT_LENGTH} > length($body)){
		return undef;
	}
	(my $cmd = $param->{PATH_INFO}) =~ s|.*/||;
	my $msg = decode_json($body);
	if(exists $command{$cmd}){
		$command{$cmd}->($fh,$msg);
		return 1;
	}else{
		send_message($fh,{status=>200,type=>"response",command=>$cmd,error=>405,text=>"invalid command",ts=>ts(),rid => -1});
		return 1;
	}
}
# отсылаем сообщение на сокет
# соообщение будет отослано как только это будет возможно
# сокет хранится только здесь
# сообщение в виде perl-структуры
sub send_message {
	my ($fh,$msg) = @_;
	my $status = delete $msg->{status} || 200;
	my $body = encode_json($msg);
	my $ret = "HTTP/1.1 ".$status." ".HTTP::Status::status_message($status)."\n";
	$ret .= "Content-Length: ".length($body)."\n";
	$ret .= "Content-Type: text/x-javascript\n";
	$ret .= "\n";
	$ret .= $body;
	my $nw = EV::io $fh, EV::WRITE, \&onwrite_handler;
	# создаем слушателя и сохраняем необходимые данные
	$writers{int $nw} = [$nw,$fh,$ret,$msg->{type}];
	return undef;
}
sub onwrite_handler {
	my ($w, $r) = @_;
	my $ary = $writers{int $w};
	my $need = length($ary->[2]);
	if($debug){
		my $buf = $ary->[2];
		#utf8::decode $buf;
		print "write: $buf\n";
	}
	my $rs;
	eval {
		$rs = $ary->[1]->send($ary->[2]);
	};
	if($@){
		print "ERROR: $!\n";
		print $ary->[2],"\n";
		print "type: $ary->[3]\n";
	}
	if($rs){
		$need -= $rs;
		# если отправлено не все, то продолжаем ждать
		if($need > 0){
			substr($ary->[2],0,$rs,"");
			if($debug){
				print "write only $rs need $need\n";
			}
			return 0;
		}
		if($debug){
			print "write ok $rs $need $ary->[3]\n";
		}
	}
	delete $writers{int $w};
}
sub generate_sid {
	return md5_hex(gettimeofday(),rand());
}
sub clear_session {
	my $login = shift;
	my $new_sid = shift || "";
	for my $ch (keys %{$logins{$login}{chanels}}){
		leave_chanel($login,$ch);
	}
	$logins{$login}{chanels} = {};
	if(my $old_sid = $logins{$login}{sid}){
		delete $sids{$old_sid};
	}
	$logins{$login}{sid} = $new_sid;
	delete $waiting{$login};
}
# обработчики команд
sub cmd_login {
	my ($fh,$msg) = @_;
	my $login;
	unless($login = $msg->{login}){
		send_message($fh,{status=>200,type=>"response",command=>"login",error=>412,text=>"need login",ts=>ts(),rid=>$msg->{rid}});
		return undef;
	}

	my $sid = generate_sid();
	if($logins{$login}){
		clear_session($login,$sid);
	}else{
		$logins{$login} = {
			sid => $sid,
			private => [],
			chanels => {},
		};
	}

	my $now = time();
	$sids{$sid} = {
		login => $login,
		start => $now,
		last => $now,
	};

	send_message($fh,{status=>200,type=>"response",command=>"login",error=>0,text=>"welcome [$login]",ts=>ts(),sid => $sid,rid=>$msg->{rid}});
}

sub cmd_message {
	my ($fh,$msg) = @_;
	my $login;
	if($login = check_auth($msg->{sid})){
		$msg->{chanel}||="";
		if(defined $logins{$login}{chanels}{$msg->{chanel}}){
			send_message($fh,{status=>200,type=>"response",command=>"message",error=>0,text=>"ok",ts=>ts(),rid=>$msg->{rid}});
			put_message({rid=>$msg->{rid},from=>$login,text=>$msg->{text},chanel=>$msg->{chanel}||""});
		}else{
			send_message($fh,{status=>200,type=>"response",command=>"message",error=>403,text=>"not authority",ts=>ts(),rid=>$msg->{rid}});
		}
	}else{
		send_message($fh,{status=>200,type=>"response",command=>"message",error=>401,text=>"not authority",ts=>ts(),rid=>$msg->{rid}});
	}
}
sub cmd_private {
	my ($fh,$msg) = @_;
	my $login;
	if($login = check_auth($msg->{sid})){
		send_message($fh,{status=>200,type=>"response",command=>"private",error=>0,text=>"ok",ts=>ts(),rid=>$msg->{rid}});
		put_private({rid=>$msg->{rid},from=>$login,text=>$msg->{text},to => $msg->{to}});
	}else{
		send_message($fh,{status=>200,type=>"response",command=>"private",error=>401,text=>"not authority",ts=>ts(),rid=>$msg->{rid}});
	}
}
# вот здесь нужно ждать
sub cmd_check {
	my ($fh,$msg) = @_;
	my $login;
	unless($login = check_auth($msg->{sid})){
		send_message($fh,{status=>200,type=>"response",command=>"check",error=>401,text=>"not authority",ts=>ts(),rid=>$msg->{rid}});
	}else{
		# получаем список сообщений
		my $list = build_list($login);
		#  если есть ли новые сообщения
		if(@$list){
			send_message($fh,{status=>200,type=>"list",list=>$list,rid=>$msg->{rid}});
		}else{
			# в список ожидающих поступления сообщений нужно сохранить сокет
			$waiting{$login} = {fh => $fh, sid => $msg->{sid},rid => $msg->{rid}};
		}
	}
}
sub cmd_list_chanel {
	my ($fh,$msg) = @_;
	my $login;
	unless($login = check_auth($msg->{sid})){
		send_message($fh,{status=>200,type=>"response",command=>"list",error=>401,text=>"not authority",ts=>ts(),rid=>$msg->{rid}});
	}elsif($msg->{chanel}){
		if(my $list = list_chanel($login,$msg->{chanel})){
			put_private({rid=>$msg->{rid},chanel=>$msg->{chanel},from=>"",text=>$list,to=>[$login]});
			send_message($fh,{status=>200,type=>"response",command=>"list",chanel=>$msg->{chanel},error=>0,text=>"ok",ts=>ts(),rid=>$msg->{rid}});
		}else{
			send_message($fh,{status=>200,type=>"response",command=>"list",chanel=>$msg->{chanel},error=>404,text=>"bad chanel",ts=>ts(),rid=>$msg->{rid}});
		}
	}else{
		send_message($fh,{status=>200,type=>"response",command=>"list",error=>412,text=>"need chanel",ts=>ts(),rid=>$msg->{rid}});
	}
}
sub cmd_join_chanel {
	my ($fh,$msg) = @_;
	my $login;
	unless($login = check_auth($msg->{sid})){
		send_message($fh,{status=>200,type=>"response",command=>"join",error=>401,text=>"not authority",ts=>ts(),rid=>$msg->{rid}});
	}elsif($msg->{chanel}){
		if(join_chanel($login,$msg->{chanel})){
			put_message({rid=>$msg->{rid},text => "joined [$login]",chanel=>$msg->{chanel}});
			send_message($fh,{status=>200,type=>"response",command=>"join",chanel=>$msg->{chanel},error=>0,text=>"ok",ts=>ts(),rid=>$msg->{rid}});
		}else{
			send_message($fh,{status=>200,type=>"response",command=>"join",chanel=>$msg->{chanel},error=>404,text=>"bad chanel",ts=>ts()});
		}
	}else{
		send_message($fh,{status=>200,type=>"response",command=>"join",error=>412,text=>"need chanel",ts=>ts(),rid=>$msg->{rid}});
	}
}
sub cmd_leave_chanel {
	my ($fh,$msg) = @_;
	my $login;
	unless($login = check_auth($msg->{sid})){
		send_message($fh,{status=>200,type=>"response",command=>"leave",error=>401,text=>"not authority",ts=>ts(),rid=>$msg->{rid}});
	}elsif($msg->{chanel}){
		if(leave_chanel($login,$msg->{chanel})){
			put_message({rid=>$msg->{rid},text => "leave [$login]",chanel=>$msg->{chanel}});
			send_message($fh,{status=>200,type=>"response",command=>"leave",chanel=>$msg->{chanel},error=>0,text=>"ok",ts=>ts(),rid=>$msg->{rid}});
		}else{
			send_message($fh,{status=>200,type=>"response",command=>"leave",chanel=>$msg->{chanel},error=>404,text=>"bad chanel",ts=>ts(),rid=>$msg->{rid}});
		}
	}else{
		send_message($fh,{status=>200,type=>"response",command=>"leave",chanel=>$msg->{chanel},error=>412,text=>"need chanel",ts=>ts(),rid=>$msg->{rid}});
	}
}
sub cmd_create_chanel {
	my ($fh,$msg) = @_;
	my $login;
	unless($login = check_auth($msg->{sid})){
		send_message($fh,{status=>200,type=>"response",command=>"create",error=>401,text=>"not authority",ts=>ts(),rid=>$msg->{rid}});
	}elsif($msg->{chanel}){
		if(add_chanel($login,$msg->{chanel})){
			send_message($fh,{status=>200,type=>"response",command=>"create",error=>0,text=>"ok",ts=>ts(),rid=>$msg->{rid}});
		}else{
			send_message($fh,{status=>200,type=>"response",command=>"create",error=>404,text=>"bad chanel",ts=>ts(),rid=>$msg->{rid}});
		}
	}else{
		send_message($fh,{status=>200,type=>"response",command=>"create",error=>412,text=>"need chanel",ts=>ts(),rid=>$msg->{rid}});
	}
}
sub cmd_remove_chanel {
	my ($fh,$msg) = @_;
	my $login;
	unless($login = check_auth($msg->{sid})){
		send_message($fh,{status=>200,type=>"response",command=>"remove",error=>401,text=>"not authority",ts=>ts(),rid=>$msg->{rid}});
	}elsif($msg->{chanel}){
		unless(my $e = del_chanel($login,$msg->{chanel})){
			send_message($fh,{status=>200,type=>"response",command=>"remove",error=>0,text=>"ok",ts=>ts(),rid=>$msg->{rid}});
		}else{
			send_message($fh,{status=>200,type=>"response",command=>"remove",error=>$e,text=>"bad chanel",ts=>ts(),rid=>$msg->{rid}});
		}
	}else{
		send_message($fh,{status=>200,type=>"response",command=>"remove",error=>412,text=>"need chanel",ts=>ts(),rid=>$msg->{rid}});
	}
}
###############
# рассылка приватного сообщения
sub put_private {
	my ($msg) = @_;
	$msg->{ts} = ts();
	$msg->{ats} = ats();
	for my $to(@{$msg->{to}}){
		# сохраняем сообщение
		if(my $l = $logins{$to}){
			push @{$l->{private}},$msg;
			# если есть в списке ожидающих, то отправляем сразу
			if(my $target = delete $waiting{$to}){
				my $list;
				if(@{$list = build_list($to)}){
					send_message($target->{fh},{status=>200,type=>"list",list=>$list,rid=>$target->{rid}});
				}
			}
		}
	}
}
# рассылка входящего сообщения
sub put_message {
	my ($msg) = @_;
	my $chanel = $chanels{$msg->{chanel}};
	unless($chanel){
		return;
	}
	$msg->{ts} = ts();
	$msg->{ats} = ats();
	# так как отдаем не больше 100 последних, то хранить больше не имеет смысл
	push @{$chanel->{messages}},$msg;
	if(@{$chanel->{messages}} > 500){
		splice @{$chanel->{messages}},0,350;
		$chanel->{shift} += 350;
	}
	$chanel->{last} = time();
	# пройти по всем ожидающим и отправить
	my @wdone;
	for my $login(keys %{$chanel->{list}}){
		if(my $w = $waiting{$login}){
			my $list;
			if(@{$list = build_list($login)}){
				send_message($w->{fh},{status=>200,type=>"list",list=>$list,rid=>$w->{rid}});
				push @wdone,$login;
			}
		}
	}
	# удаляем всех кому отправили
	delete @waiting{@wdone};
}
sub build_list {
	my ($login) = @_;
	unless($login){
		# если сессии уже нет, то ничего не делаем
		return [];
	}
	my @list;

	my $lgn = $logins{$login};
	if(@{$lgn->{private}}){
		push @list,@{$lgn->{private}};
		$lgn->{private} = [];
	}
	for my $id(keys %{$lgn->{chanels}}){
		my $ch = $chanels{$id};
		next unless $ch;

		my $last = $lgn->{chanels}{$id};
		my $messages = $ch->{messages};
		# отсылаем только последние 100 сообщений
		if(@$messages - ($last -= $ch->{shift}) > 100){
			$last = @$messages - 100;
		}
		if($last < @$messages){
			push @list,@{$messages}[$last .. $#$messages];
			$lgn->{chanels}{$id} = $ch->{shift} + @$messages;
		}
	}
	return \@list;
}
#####################################

sub check_list_handler {
	my ($w) = @_;
	my $ch = $chanel_lister{int $w};
	check_list($ch);
}
sub check_list {
	my $ch = shift;
	my @delete;
	my $sharp = time() - $ttl;
	for my $login(keys %{$chanels{$ch}->{list}}){
		my $last = ( $sids{ ($logins{$login}||{})->{sid} || "" } || {} )->{last} || 0;
		if($last < $sharp){
			push @delete,$login;
			if($logins{$login}){
				delete $logins{$login}{$ch};
			}
		}
	}
	if(@delete){
		delete @{$chanels{$ch}->{list}}{@delete};
	}
	$chanels{$ch}->{list_str} = join(",",sort {$a cmp $b} keys %{$chanels{$ch}{list}});
}
sub list_chanel {
	my ($owner,$id) = @_;
	unless($chanels{$id}){
		return "";
	}elsif($chanels{$id}{list}{$owner}){
		if($chanels{$id}{nolist}){
			return "no list for $id";
		}else{
			return "[".$chanels{$id}->{list_str}."]";
		}
	}else{
		return "";
	}
}
sub add_chanel {
	my ($owner,$id,%param) = @_;
	if($chanels{$id}){
		return 0;
	}else{
		my $w = $param{nolist} ? undef : EV::timer 15, 15, \&check_list_handler;
		$chanel_lister{int $w} = $id if $w;
		$chanels{$id} = {
			nolist => $param{nolist},
			lister => $w,
			owner => $owner,
			messages => [],
			shift => 0,
			list => {},
			list_str => "",
			create => time(),
			last => time(),
		};
		return 1;
	}
}
sub del_chanel {
	my ($owner,$id) = @_;
	unless($chanels{$id}){
		return 404;
	}elsif($chanels{$id}{owner} eq $owner){
		my $ch = delete $chanels{$id};
		$ch->{lister}->stop if $ch->{lister};
		for my $u(@{$ch->{list}}){
			delete $logins{$u}{chanels}{$id};
		}
		delete $chanel_lister{int $ch->{lister}} if $ch->{lister};
		$ch->{messages} = undef;
		$ch->{list} = undef;
		$ch->{lister} = undef;
		return 0;
	}else{
		return 403;
	}
}
sub join_chanel{
	my($login,$id) = @_;
	if(my $c = $chanels{$id}){
		$c->{list}{$login} = 1;
		$logins{$login}{chanels}{$id} = $c->{shift} + @{$c->{messages}};
		$c->{list_str} .= ",".$login unless $c->{nolist};
		return 1;
	}else{
		return 0;
	}
}
sub leave_chanel{
	my($login,$id) = @_;
	if(my $c = $chanels{$id}){
		delete $c->{list}{$login};
		delete $logins{$login}{chanels}{$id};
		return 1;
	}else{
		return 0;
	}
}
